package com.hopu.shop;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.shop.CateViewTop3;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

//未写入到数据库
public class CateViewTop3Ana {

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<CateViewTop3, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String cid = json.getString("category_id");
            String cname = json.getString("category_name");
            String type = json.getString("event_type");

            return new Tuple2<>(new CateViewTop3(cid, cname, type, 0), 1);
        });

        JavaPairRDD<CateViewTop3, Integer> viewRDD = pairRDD.filter(t -> "view".equals(t._1.getType()));

        JavaPairRDD<CateViewTop3, Iterable<Integer>> group = viewRDD.groupByKey();

        viewRDD = group.mapToPair(t -> {
            CateViewTop3 cateViewTop3 = t._1;
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            cateViewTop3.setCount(count);
            return new Tuple2<>(cateViewTop3, count);
        });

        JavaPairRDD<CateViewTop3, Integer> sortByKey = viewRDD.sortByKey();
        List<CateViewTop3> top3s = sortByKey.map(t -> t._1).take(3);
        top3s.forEach(System.out::println);

        for (CateViewTop3 top3: top3s) {
            String[] columns = {"cid", "cname", "count"};
            String[] values = {top3.getCid(), top3.getCname(), top3.getCount()+""};
            try {
                HBaseUtils.putOneRowToHbase("shopana", top3.getCid(), "top3view", columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}
